package com.firewolf.rocketmq;

import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * 作者：刘兴 时间:2019-07-11
 **/
public class Consumer {

  public static void main(String[] args) throws InterruptedException, MQClientException {

    //创建消费者实例，指定消费者组
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group1");

    //指定nameserver地址，多个用;隔开
    consumer.setNamesrvAddr("10.211.55.6:9876");

    //指定订阅的主题和tag，*表示订阅这个主题下的所有消息
    consumer.subscribe("mytopic", "*");

    //注册接受到消息的回调函数
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
          ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : msgs) {
          System.out.printf("接受消息：%s %n", new String(msg.getBody()));
        }
//        System.out.printf("%s 接收到消息: %s %n", Thread.currentThread().getName(), msgs);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });

    //Launch the consumer instance.
    consumer.start();

    System.out.printf("消费者启动了...%n");
  }
}